#include "StdAfx.h"
#include "RangeWindowOperator.h"
#include "QueueEntity.h"

RangeWindowOperator::RangeWindowOperator(Timeunit timeRange):rangeWindowSize(timeRange)
{
	this->windowSynopsis.reset(new WindowSynopsis());

	this->lastInputTimestamp = 0;
}


RangeWindowOperator::~RangeWindowOperator(void)
{
}


void RangeWindowOperator::execution(int executionTimes)
{
#ifdef DEBUG
	std::cout<<"===================operator begin================="<<std::endl;
	std::cout<<"operatorid : "<<this->getId()<<std::endl;
#endif
	assert(this->getInputQueueList().size()==1);
	assert(this->getOutputQueueList().size()==1);
	boost::shared_ptr<QueueEntity>inputQueue = this->getInputQueueList().front();
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	if(this->getOperatorState()==BLOCKED)
	{
		dealWithBlock();
		if(this->getOperatorState() == BLOCKED)
		{
			return;
		}
	}
	for(int i = 0; i < executionTimes ; i++)
	{
		Element inputElement;

		if(inputQueue->isEmpty())   // the input queue is not empty
		{
			break;
		}
		if(outputQueue->isFull())   // the output queue is not full 
		{
			break;
		}
		inputQueue->dequeue(inputElement);


		assert(inputElement.mark==PLUS_MARK);
		assert(lastInputTimestamp<= inputElement.timestamp);
		lastInputTimestamp = inputElement.timestamp;

		if(this->windowSynopsis->isFull())   // the window synopsis is not full
		{
			throw runtime_error("synopsis is full");
		}

		this->windowSynopsis->insertElement(inputElement);
	

		if(lastInputTimestamp>=this->rangeWindowSize)
		{  // generate oldest element out
			expireElement( lastInputTimestamp - rangeWindowSize); //expire tuples with timestamp <= lastInputTimestamp - rangeWindowSize	
		}
		if(this->getOperatorState()==BLOCKED)
		{
			blockedElement = inputElement;
			return;
		}
		if(outputQueue->isFull())
		{
			this->setOperatorState(BLOCKED);
			blockedElement = inputElement;
			return;
		}
		outputQueue->enqueue(inputElement);//forward the same element
	
	}
#ifdef DEBUG
	std::cout<<"===================operator over================="<<std::endl;
#endif
}
void RangeWindowOperator::expireElement(Timestamp ts)//expire elements with a timestamp older than ts 
{
	
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	Element elementInSynopsis;
	Element elementOutput;
	while(true)
	{
		if (this->windowSynopsis -> isEmpty())
		{
			return;
		}
		this->windowSynopsis->getOldestElement(elementInSynopsis);
		if(elementInSynopsis.timestamp>ts)
		{
			break;
		}
		if(outputQueue->isFull())
		{
			 this->setOperatorState(BLOCKED);
			 return;
		}
		elementOutput.document = elementInSynopsis.document;
		elementOutput.id = elementInSynopsis.id;
		elementOutput.mark = MINUS_MARK;
		elementOutput.timestamp = elementInSynopsis.timestamp + rangeWindowSize;

		outputQueue->enqueue(elementOutput);
		windowSynopsis->deleteOldestElement();
	}
}
void RangeWindowOperator::dealWithBlock()
{
#ifdef DEBUG
	std::cout<<"**********deal with block begin************"<<std::endl;
#endif
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	assert(this->getOperatorState() == BLOCKED);
	this->setOperatorState(NORMAL);
	if(lastInputTimestamp>=this->rangeWindowSize)
	{
		expireElement(lastInputTimestamp-rangeWindowSize);
	}
	if(this->getOperatorState()==BLOCKED)
	{
		return;
	}
	if(outputQueue->isFull())
	{
		this->setOperatorState(NORMAL);
		return;
	}
	outputQueue->enqueue(blockedElement);
#ifdef DEBUG
	std::cout<<"**********deal with block over************"<<std::endl;
#endif
}